package jk.fox.engine;

import jk.fox.common.payload.container.RpcContainer;
import jk.fox.common.payload.message.MessageDispatcher;
import jk.fox.config.RpcConfiguration;
import jk.fox.engine.listener.RpcServiceRegistryListener;
import jk.fox.engine.listener.RpcServiceStartedListener;
import jk.fox.engine.context.AbstractInvocationRpcContext;
import jk.fox.engine.context.DefaultRpcContext;
import jk.fox.engine.context.RpcContext;
import jk.fox.engine.process.RpcContextAware;
import jk.fox.exchange.Exchange;
import jk.fox.exchange.server.ServerExchange;
import jk.fox.exchange.client.ClientExchange;
import jk.fox.exchange.server.handler.StandardExchangeHandler;
import jk.fox.node.client.RpcClient;
import jk.fox.node.client.RpcCluster;
import jk.fox.node.rule.RuleFactory;
import jk.fox.registry.Registry;
import jk.fox.registry.discovery.NodeDiscovery;
import jk.fox.registry.register.NodeRegister;
import jk.fox.transport.ClientTransporter;
import jk.fox.transport.ServerTransporter;
import jk.fox.transport.Transport;

import java.util.ArrayList;
import java.util.List;

/**
 * @author cuichao
 * @Description: DefaultRpcEngine
 * @Date: create in 2020/5/31 22:33
 */
public class DefaultRpcEngine implements RpcEngine {
    /**
     * Rpc配置
     */
    private RpcConfiguration configure;
    /**
     * Rpc上下文
     */
    private RpcContext context;
    /**
     * Rpc节点容器
     */
    private RpcContainer container;
    /**
     * Rpc传输器模块
     */
    private Transport transporter;
    /**
     * Rpc传输交换器模块
     */
    private Exchange exchange;
    /**
     * Rpc 注册发现模块
     */
    private Registry registry;
    /**
     * Rpc消息分发器
     */
    private MessageDispatcher dispatcher;
    /**
     * Rpc负载均衡工厂
     */
    private RuleFactory ruleFactory;
    /**
     * Rpc启动监听器
     */
    private List<RpcServiceStartedListener> startedListeners = new ArrayList<>();

    private List<RpcContextAware> processors = new ArrayList<>();

    public DefaultRpcEngine (RpcConfiguration config){
        this (config,null);
    }

    public DefaultRpcEngine (RpcConfiguration config , RpcContext rpcContext){
        this.configure = config;
        this.container = config.getContainer();
        if(rpcContext == null){
            this.context = new DefaultRpcContext(container);
        }else{
            this.context = rpcContext;
        }
        initWithConfig(config);
    }

    protected void initWithConfig(RpcConfiguration config){
        this.ruleFactory = config.getRuleFactory();
        this.dispatcher =  config.getMessageDispatcher();
        this.transporter = config.getTransportConfig().create(dispatcher);
        this.exchange = config.getExchangeConfig().create(dispatcher);
        this.registry = config.getRegistryConfig().create(null);

    }



    @Override
    public void registry(Class target) {
        this.context.registryReference(target.getName(),target,null);
    }

    @Override
    public void registry(Class target, Object object) {
        this.context.registryService(target.getName(),target,object);
    }


    @Override
    public Object getConsumer(String key) {
        return this.container.getReference(key);
    }

    @Override
    public Object getProvider(String key) {
        return  this.container.getService(key);
    }

    @Override
    public RpcNodeDefinition getDefinition(String key) {
        return this.context.getRpcNodeDefinition(key);
    }

    /**
     * 客户端启动
     */
    private void bootstrapClient() throws Exception{
        //客户端传输器
        ClientTransporter clientTransport = transporter.client();
        //客户端交换器
        ClientExchange clientExchange = exchange.client(clientTransport);
        //节点发现
        NodeDiscovery discovery = registry.discovery();
        //节点客户端
        RpcCluster cluster = new RpcClient(discovery, ruleFactory, clientExchange, configure);
        if(!(context instanceof AbstractInvocationRpcContext)){
            throw new ClassCastException(String.format("%s must be extends %s", context.getClass().getName(),AbstractInvocationRpcContext.class.getName()));
        }
        //客户端端消息分发器 与 交互器、传输器连接
        dispatcher.clientExchange(clientExchange);
        dispatcher.clientTransporter(clientTransport);
        //上层API执行器连节点客户端
        AbstractInvocationRpcContext context = (AbstractInvocationRpcContext) this.context;
        context.invocation().cluster(cluster);
    }

    /**
     * 服务端启动
     */
    private void bootstrapServer() throws Exception{
        //服务端传输器
        ServerTransporter server = transporter.server();
        //服务端执行器连接上层API执行器
        if(!(context instanceof AbstractInvocationRpcContext)){
            throw new ClassCastException(String.format("%s must be extends %s", context.getClass().getName(),AbstractInvocationRpcContext.class.getName()));
        }
        AbstractInvocationRpcContext context = (AbstractInvocationRpcContext) this.context;
        StandardExchangeHandler exchangeHandler = new StandardExchangeHandler(context.invocation());
        //服务端交换器
        ServerExchange serverExchange = exchange.server(server,exchangeHandler);
        //服务端消息分发器 与 交互器、传输器连接
        dispatcher.serverExchange(serverExchange);
        dispatcher.serverTransporter(server);
        //注解Rpc服务节点
        NodeRegister register = registry.register();
        start(register,serverExchange);
    }


    /**
     * 启动Rpc服务
     * @param register
     * @param exchange
     * @throws Exception
     */
    private void start(NodeRegister register , ServerExchange exchange) throws Exception{
        //添加监听
        List<RpcServiceStartedListener> listeners = buildRpcServiceStartedListener(register);
        for (RpcServiceStartedListener listener : listeners) {
            exchange.addListener(listener);
        }
        int expose = configure.getExpose();
        if(expose < 1){
            throw new RuntimeException("the rpcServer is not bind port");
        }
        //call rpcContext-aware
        for (RpcContextAware processor : processors) {
            processor.setRpcContext(this.context);
        }
        expose(exchange,expose);
    }

    /**
     * 暴露Rpc服务端口
     * @param exchange
     * @param port
     */
    private void expose(ServerExchange exchange,int port){
        new Thread(() -> {
            try {
                exchange.bind(port);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }).start();
    }

    /**
     * 构建Rpc服务启动监听器
     * @return
     */
    private List<RpcServiceStartedListener> buildRpcServiceStartedListener(NodeRegister register){
        List<RpcServiceStartedListener> listeners = new ArrayList<>();
        List<RpcServiceStartedListener> interiors = interiorRpcServiceStartedListener(register);
        listeners.addAll(interiors);
        if(!startedListeners.isEmpty()){
            listeners.addAll(startedListeners);
        }
        for (RpcServiceStartedListener listener : listeners) {
            deductionRpcContextAware(listener);
        }

        return listeners;
    }

    /**
     * 内部启动监听器
     * @param register
     * @return
     */
    private List<RpcServiceStartedListener> interiorRpcServiceStartedListener(NodeRegister register){
        List<RpcServiceStartedListener> listeners = new ArrayList<>();
        RpcServiceRegistryListener registryListener = new RpcServiceRegistryListener(register);
        listeners.add(registryListener);
        return listeners;
    }


    /**
     * 推断是否为感知处理器子类
     * @param instance
     */
    private void deductionRpcContextAware(Object instance){
        if(instance instanceof RpcContextAware){
            processors.add((RpcContextAware) instance);
        }
    }

    @Override
    public void start() throws Exception {
        //服务端启动
        bootstrapServer();
        //客户端启动
        bootstrapClient();
    }

    @Override
    public void addRpcServiceStartedListener(RpcServiceStartedListener listener) {
        this.startedListeners.add(listener);
    }


}
